#!/usr/bin/python
# -*- coding: utf-8 -*-
# @Version        : 1.0
# @Update Time    : 2025/8/26 10:00
# @File           : file_utils.py
# @IDE            : PyCharm
# @Desc           : 文件处理工具类，提供文件类型检测、缩略图生成、文件元数据提取等功能

import hashlib
import mimetypes
import os
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union

from fastapi import UploadFile
from applications.core.exceptions import BaseAppException

# PIL/Pillow 可选导入
try:
    from PIL import Image, ImageOps  # type: ignore

    HAS_PIL = True
except ImportError:
    HAS_PIL = False
    Image = None
    ImageOps = None


class FileUtils:
    """文件处理工具类

    提供文件类型检测、缩略图生成、文件元数据提取等功能
    """

    # 图片格式
    IMAGE_FORMATS = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp", ".tiff", ".ico"}

    # 视频格式
    VIDEO_FORMATS = {".mp4", ".avi", ".mov", ".wmv", ".flv", ".webm", ".mkv", ".m4v"}

    # 音频格式
    AUDIO_FORMATS = {".mp3", ".wav", ".flac", ".aac", ".ogg", ".wma", ".m4a"}

    # 文档格式
    DOCUMENT_FORMATS = {".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx", ".txt", ".rtf"}

    # 压缩包格式
    ARCHIVE_FORMATS = {".zip", ".rar", ".7z", ".tar", ".gz", ".bz2"}

    @classmethod
    def get_file_type_by_extension(cls, filename: str) -> str:
        """通过文件扩展名获取文件类型

        Args:
            filename: 文件名

        Returns:
            str: 文件类型 (image, video, audio, document, archive, other)
        """
        ext = Path(filename).suffix.lower()

        if ext in cls.IMAGE_FORMATS:
            return "image"
        elif ext in cls.VIDEO_FORMATS:
            return "video"
        elif ext in cls.AUDIO_FORMATS:
            return "audio"
        elif ext in cls.DOCUMENT_FORMATS:
            return "document"
        elif ext in cls.ARCHIVE_FORMATS:
            return "archive"
        else:
            return "other"

    @classmethod
    def get_file_type_by_content(cls, file_path: str) -> str:
        """通过文件扩展名和mimetypes获取MIME类型

        Args:
            file_path: 文件路径

        Returns:
            str: MIME类型
        """
        # 使用mimetypes库（跨平台兼容）
        mime_type, _ = mimetypes.guess_type(file_path)
        if mime_type:
            return mime_type

        # 备选方案：根据文件扩展名推断
        ext = Path(file_path).suffix.lower()
        mime_map = {
            # 图片类型
            ".jpg": "image/jpeg",
            ".jpeg": "image/jpeg",
            ".png": "image/png",
            ".gif": "image/gif",
            ".webp": "image/webp",
            ".bmp": "image/bmp",
            ".tiff": "image/tiff",
            ".ico": "image/x-icon",
            # 文档类型
            ".pdf": "application/pdf",
            ".txt": "text/plain",
            ".doc": "application/msword",
            ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
            ".xls": "application/vnd.ms-excel",
            ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            ".ppt": "application/vnd.ms-powerpoint",
            ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
            # 压缩文件
            ".zip": "application/zip",
            ".rar": "application/x-rar",
            ".7z": "application/x-7z-compressed",
            ".tar": "application/x-tar",
            ".gz": "application/gzip",
            # 视频类型
            ".mp4": "video/mp4",
            ".avi": "video/x-msvideo",
            ".mov": "video/quicktime",
            ".wmv": "video/x-ms-wmv",
            ".flv": "video/x-flv",
            ".webm": "video/webm",
            ".mkv": "video/x-matroska",
            # 音频类型
            ".mp3": "audio/mpeg",
            ".wav": "audio/wav",
            ".flac": "audio/flac",
            ".aac": "audio/aac",
            ".ogg": "audio/ogg",
            ".wma": "audio/x-ms-wma",
            ".m4a": "audio/mp4",
        }
        return mime_map.get(ext, "application/octet-stream")

    @classmethod
    def validate_file_extension(cls, filename: str, allowed_extensions: List[str]) -> bool:
        """验证文件扩展名

        Args:
            filename: 文件名
            allowed_extensions: 允许的扩展名列表

        Returns:
            bool: 验证结果
        """
        ext = Path(filename).suffix.lower()
        return ext in [ext.lower() for ext in allowed_extensions]

    @classmethod
    def calculate_file_hash(cls, file_path: str, algorithm: str = "md5") -> str:
        """计算文件哈希值

        Args:
            file_path: 文件路径
            algorithm: 哈希算法 (md5, sha1, sha256)

        Returns:
            str: 文件哈希值
        """
        hash_algorithms = {"md5": hashlib.md5(), "sha1": hashlib.sha1(), "sha256": hashlib.sha256()}

        if algorithm not in hash_algorithms:
            raise ValueError(f"不支持的哈希算法: {algorithm}")

        hasher = hash_algorithms[algorithm]

        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hasher.update(chunk)

        return hasher.hexdigest()

    @classmethod
    def get_file_info(cls, file_path: str) -> Dict:
        """获取文件详细信息

        Args:
            file_path: 文件路径

        Returns:
            Dict: 文件信息字典
        """
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"文件不存在: {file_path}")

        path_obj = Path(file_path)
        stat = path_obj.stat()

        file_info = {
            "filename": path_obj.name,
            "extension": path_obj.suffix.lower(),
            "size": stat.st_size,
            "size_human": cls.format_file_size(stat.st_size),
            "created_time": datetime.fromtimestamp(stat.st_ctime),
            "modified_time": datetime.fromtimestamp(stat.st_mtime),
            "accessed_time": datetime.fromtimestamp(stat.st_atime),
            "file_type": cls.get_file_type_by_extension(path_obj.name),
            "mime_type": cls.get_file_type_by_content(file_path),
            "md5": cls.calculate_file_hash(file_path, "md5"),
        }

        # 如果是图片且PIL可用，添加图片信息
        if file_info["file_type"] == "image" and HAS_PIL:
            image_info = cls.get_image_info(file_path)
            file_info.update(image_info)

        return file_info

    @classmethod
    def format_file_size(cls, size_bytes: int) -> str:
        """格式化文件大小

        Args:
            size_bytes: 文件大小（字节）

        Returns:
            str: 格式化后的文件大小
        """
        if size_bytes == 0:
            return "0B"

        size_names = ["B", "KB", "MB", "GB", "TB"]
        i = 0
        size = size_bytes

        while size >= 1024 and i < len(size_names) - 1:
            size /= 1024.0
            i += 1

        return f"{size:.2f}{size_names[i]}"

    @classmethod
    def get_image_info(cls, image_path: str) -> Dict:
        """获取图片信息

        Args:
            image_path: 图片路径

        Returns:
            Dict: 图片信息字典
        """
        if not HAS_PIL or not Image:
            return {}

        try:
            with Image.open(image_path) as img:
                return {
                    "width": img.width,
                    "height": img.height,
                    "format": img.format,
                    "mode": img.mode,
                    "has_transparency": img.mode in ("RGBA", "LA") or "transparency" in img.info,
                }
        except Exception:
            return {}

    @classmethod
    def create_thumbnail(
        cls, image_path: str, thumbnail_path: str, size: Tuple[int, int] = (200, 200), quality: int = 85
    ) -> bool:
        """创建缩略图

        Args:
            image_path: 原图路径
            thumbnail_path: 缩略图保存路径
            size: 缩略图尺寸
            quality: 图片质量 (1-100)

        Returns:
            bool: 创建结果
        """
        if not HAS_PIL or not Image:
            raise BaseAppException(msg="PIL库未安装，无法创建缩略图")

        try:
            with Image.open(image_path) as img:
                # 创建缩略图目录
                Path(thumbnail_path).parent.mkdir(parents=True, exist_ok=True)

                # 使用兼容的缩放算法
                try:
                    # 新版本PIL使用Resampling.LANCZOS
                    resample = Image.Resampling.LANCZOS  # type: ignore
                except AttributeError:
                    # 旧版本PIL使用LANCZOS
                    resample = getattr(Image, "LANCZOS", 1)  # type: ignore

                img.thumbnail(size, resample)  # type: ignore

                # 如果是RGBA模式，转换为RGB
                if img.mode == "RGBA":
                    background = Image.new("RGB", img.size, (255, 255, 255))
                    background.paste(img, mask=img.split()[-1])
                    img = background

                # 保存缩略图
                img.save(thumbnail_path, format="JPEG", quality=quality, optimize=True)
                return True
        except Exception as e:
            raise BaseAppException(msg=f"创建缩略图失败: {str(e)}")

    @classmethod
    def compress_image(
        cls,
        image_path: str,
        output_path: Optional[str] = None,
        quality: int = 85,
        max_width: Optional[int] = None,
        max_height: Optional[int] = None,
    ) -> str:
        """压缩图片

        Args:
            image_path: 原图路径
            output_path: 输出路径，如果为None则覆盖原文件
            quality: 图片质量 (1-100)
            max_width: 最大宽度
            max_height: 最大高度

        Returns:
            str: 压缩后的文件路径
        """
        if not HAS_PIL or not Image:
            raise BaseAppException(msg="PIL库未安装，无法压缩图片")

        if output_path is None:
            output_path = image_path

        try:
            with Image.open(image_path) as img:
                # 如果指定了最大尺寸，进行缩放
                if max_width or max_height:
                    try:
                        # 新版本PIL使用Resampling.LANCZOS
                        resample = Image.Resampling.LANCZOS  # type: ignore
                    except AttributeError:
                        # 旧版本PIL使用LANCZOS
                        resample = getattr(Image, "LANCZOS", 1)  # type: ignore

                    img.thumbnail((max_width or img.width, max_height or img.height), resample)  # type: ignore

                # 如果是RGBA模式，转换为RGB
                if img.mode == "RGBA":
                    background = Image.new("RGB", img.size, (255, 255, 255))
                    background.paste(img, mask=img.split()[-1])
                    img = background

                # 创建输出目录
                Path(output_path).parent.mkdir(parents=True, exist_ok=True)

                # 保存压缩后的图片
                img.save(output_path, format="JPEG", quality=quality, optimize=True)
                return output_path
        except Exception as e:
            raise BaseAppException(msg=f"压缩图片失败: {str(e)}")

    @classmethod
    async def validate_upload_file(
        cls,
        file: UploadFile,
        max_size: int = 10,
        allowed_extensions: Optional[List[str]] = None,
        allowed_mime_types: Optional[List[str]] = None,
    ) -> Dict:
        """验证上传文件

        Args:
            file: 上传的文件
            max_size: 最大文件大小(MB)
            allowed_extensions: 允许的扩展名列表
            allowed_mime_types: 允许的MIME类型列表

        Returns:
            Dict: 验证结果和文件信息
        """
        # 验证文件名
        if not file.filename:
            raise BaseAppException(msg="文件名不能为空")

        # 验证文件扩展名
        if allowed_extensions:
            if not cls.validate_file_extension(file.filename, allowed_extensions):
                raise BaseAppException(msg=f"不支持的文件类型，允许的类型: {', '.join(allowed_extensions)}")

        # 验证MIME类型
        if allowed_mime_types:
            if file.content_type not in allowed_mime_types:
                raise BaseAppException(msg=f"不支持的文件类型，允许的类型: {', '.join(allowed_mime_types)}")

        # 验证文件大小
        if file.size is not None:
            size_mb = file.size / 1024 / 1024
            if size_mb > max_size:
                raise BaseAppException(msg=f"文件太大，最大允许 {max_size}MB")
        else:
            # 如果无法获取文件大小，读取内容检查
            content = await file.read()
            size_mb = len(content) / 1024 / 1024
            await file.seek(0)  # 重置文件指针
            if size_mb > max_size:
                raise BaseAppException(msg=f"文件太大，最大允许 {max_size}MB")

        # 验证文件不为空
        if file.size == 0:
            raise BaseAppException(msg="文件不能为空")

        return {
            "filename": file.filename,
            "size": file.size,
            "size_mb": size_mb,
            "content_type": file.content_type,
            "file_type": cls.get_file_type_by_extension(file.filename),
        }

    @classmethod
    def clean_filename(cls, filename: str) -> str:
        """清理文件名，移除特殊字符

        Args:
            filename: 原始文件名

        Returns:
            str: 清理后的文件名
        """
        import re

        # 移除或替换特殊字符
        filename = re.sub(r'[<>:"/\\|?*]', "_", filename)
        # 移除控制字符
        filename = re.sub(r"[\x00-\x1f\x7f-\x9f]", "", filename)
        # 限制文件名长度
        name, ext = os.path.splitext(filename)
        if len(name) > 100:
            name = name[:100]

        return name + ext

    @classmethod
    def is_safe_path(cls, path: str, base_path: str) -> bool:
        """检查路径是否安全（防止路径遍历攻击）

        Args:
            path: 要检查的路径
            base_path: 基础路径

        Returns:
            bool: 路径是否安全
        """
        try:
            # 规范化路径
            path = os.path.normpath(path)
            base_path = os.path.normpath(base_path)

            # 转换为绝对路径
            abs_path = os.path.abspath(os.path.join(base_path, path))
            abs_base = os.path.abspath(base_path)

            # 检查路径是否在基础路径内
            return abs_path.startswith(abs_base)
        except Exception:
            return False
